#include "rxcpp/rx.hpp"
#include <iostream>
#include "SubscribeOn.hpp"
#include <folly/concurrency/ConcurrentHashMap.h>

#include <gtest/gtest.h>
#include "option.hpp"
#include "leveldb/db.h"
#include <version>

namespace Rx {
using namespace rxcpp;
using namespace rxcpp::sources;
using namespace rxcpp::operators;
using namespace rxcpp::util;
}

// 范围循环
void observableRange() {
  //------------- Create an Observable.. a stream of numbers
  auto observable = rxcpp::observable<>::range(1, 12);


  //------------ Subscribe (only OnNext and OnCompleted Lambda given
  observable.
	  subscribe(

	  [](int v) { printf("OnNext: %d\n", v); },

	  []() { printf("OnCompleted\n"); });
}

// 过滤
void observableFilter() {

  auto values = rxcpp::observable<>::range(1, 12)
	  .filter([](int v) {
		return v % 2 == 0;
	  })
	  .map([](int x) { return x * x; });

  values
	  .subscribe(
		  [](int v) { printf("OnNext: %d\n", v); },
		  []() { printf("OnCompleted\n"); });
}

// 从容器中取值
void iterateStl() {
  std::array<int, 3> a = {{1, 2, 3}};
  auto values1 = rxcpp::observable<>::iterate(a);

  values1.subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },
	  []() { printf("OnCompleted\n"); });
}

// 创建观察者对象
void createObserver() {

  auto init = rxcpp::observable<>::create<int>([](const rxcpp::subscriber<int> &s) {
	s.on_next(1);
	s.on_next(4);
	s.on_next(9);
	s.on_completed();
  });

  init.subscribe([](int v) { printf("OnNext: %d\n", v); },
				 []() { printf("OnCompleted\n"); });
}

// concat 顺序执行 连接可观察到的流,  Observer1结束后执行Observer2
void concatObserver() {
  auto o1 = rxcpp::observable<>::range(1, 3);

  auto o2 = rxcpp::observable<>::range(4, 6);

  auto values = o1.concat(o2);

  values.
	  subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },

	  []() { printf("OnCompleted\n"); });

}

// merge 交替执行 连接可观察到的流
// startwith操作符类似于'concat'，但是它是插入到前面
void mergeObserver() {
  auto o1 = rxcpp::observable<>::range(1, 3);

  auto o2 = rxcpp::observable<>::range(4, 6);

  auto values = o1.merge(o2);

  values.
	  subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },

	  []() { printf("OnCompleted\n"); });
}

// 发射指定值的Observable
void justObserver() {
  auto values = rxcpp::observable<>::just(1);

  values.
	  subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },
	  []() { printf("OnCompleted\n"); });
}

// 只发射前面的N项数据
void takeObserver() {
  auto values = rxcpp::observable<>::range(1); // infinite (until overflow) stream of integers

  auto s1 = values.
	  take(3).
	  map([](int prime) { return std::make_tuple("1:", prime); });

  auto s2 = values.
	  take(3).
	  map([](int prime) { return std::make_tuple("2:", prime); });

  s1
	  .concat(s2)
	  .subscribe(rxcpp::util::apply_to(
		  [](const char *s, int p) {
			printf("%s %d\n", s, p);
		  }));

  // std::tuple 0~任意个元组
  s1
	  .concat(s2)
	  .subscribe(
		  [](std::tuple<const char *, int> p) {
			printf("%s, %d\n", std::get<0>(p), std::get<1>(p));
		  });
}

// 取消订阅
void unsubscribe() {
  auto subs = rxcpp::composite_subscription();
  auto values = rxcpp::observable<>::range(1, 10);

  values.subscribe(
	  subs,
	  [&subs](int v) {
		printf("OnNext: %d\n", v);
		if (v == 6) {
		  //-- Stop recieving events
		  subs.unsubscribe();
		}
	  },
	  []() { printf("OnCompleted\n"); });
}

// map
void mapLaunch() {
  auto ints = rxcpp::observable<>::range(1, 10).
	  map([](int n) { return n * n; });

  ints.subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },
	  []() { printf("OnCompleted\n"); });
}

void average() {
  auto values = rxcpp::observable<>::range(1, 20).average();

  values.
	  subscribe(
	  [](double v) { printf("average: %lf\n", v); },
	  []() { printf("OnCompleted\n"); });
}

// 扫描操作符，每 range循环一次，计算一次scan ,seed是上次循环计算的种子值

// accumulator 蓄能器  || 累计下注(每赢一次即押于下一轮赌博)
void scan() {
  int count = 0;
  auto values = rxcpp::observable<>::range(1, 20).
	  scan(
	  0,
	  [&count](int seed, int v) {
		count++;
		return seed + v;
	  });
  values.subscribe(
	  [&](int v) { printf("Average through Scan: %f\n", (double) v / count); },
	  []() { printf("OnCompleted\n"); });
}

// 管道组合运算符
void pipeline() {
  using namespace Rx;
  auto ints = rxcpp::observable<>::range(1, 10) |
	  map([](int n) { return n * n; });
  ints.subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },
	  []() { printf("OnCompleted\n"); });

  //等价于

  // auto ints = rxcpp::observable<>::range(1, 10);
  // auto intsFromMap = ints.map([](int n) {return n * n; });
}

/**
 * @brief 支持的调度类型
 * ImmediateScheduler
 * EventLoopScheduler
 *
 * 调度器 "Scheduler"来管理多线程环境中Observable
 *
 * 多线程运行
 */
void observerOn() {

  //---------------- Generate a range of values
  //---------------- Apply Square function
  auto values = rxcpp::observable<>::range(1, 4).
	  map([](int v) {
	return v * v;
  });

  //------------- Emit the current thread details
  std::cout << "Main Thread id => "
			<< std::this_thread::get_id()
			<< std::endl;
  //---------- observe_on another thread....

  //---------- make it blocking to
  values.observe_on(rxcpp::synchronize_new_thread())
	  .as_blocking()
	  .subscribe(
		  [](int v) {
			std::cout << "Observable Thread id => "
					  << std::this_thread::get_id()
					  << "  " << v << std::endl;
		  },
		  []() { std::cout << "OnCompleted" << std::endl; });

  //------------------ Print the main thread details
  std::cout << "Main Thread id => "
			<< std::this_thread::get_id()
			<< std::endl;
}

// FlatMap 无序的，concatmap严格的顺序
void FlatMap() {
  std::array<std::string, 4> a = {{"Praseed", "Peter", "Sanjay", "Raju"}};
  auto values = rxcpp::observable<>::iterate(a).flat_map(
	  [](const std::string &v) {
		std::array<std::string, 3> salutation = {{"Mr.", "Monsieur", "Sri"}};
		return rxcpp::observable<>::iterate(salutation);
	  },
	  [](const std::string &f, const std::string &s) {
		return f + " " + s;
	  });
  values.subscribe([](const std::string &f) {
	std::cout << "f  == " << f << std::endl;
  }, []() { std::cout << "Hello World.." << std::endl; });
}

// 返回与源可观察相同的可观察值，可用于执行副作用
// 拷贝一份observable，同时执行
void tap() {
  //---- Create a mapped Observable
  auto ints = rxcpp::observable<>::range(1, 3).
	  map([](int n) { return n * n; });
  //---- Apply the tap operator...The Operator
  //---- will act as a filter/debug operator
  auto values = ints.
	  tap(
	  [](int v) { printf("Tap -       OnNext: %d\n", v); },
	  []() {
		printf("Tap -       OnCompleted\n");
	  });
  //------- Do some action
  values.
	  subscribe(
	  [](int v) { printf("Subscribe - OnNext: %d\n", v); },
	  []() { printf("Subscribe - OnCompleted\n"); });
}

// defer延迟  直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable
void defer() {
  auto observable_factory = []() {
	return rxcpp::observable<>::range(1, 3).
		map([](int n) { return n * n; });
  };
  auto init = rxcpp::observable<>::defer(observable_factory);
  init.
	  subscribe(
	  [](int v) { printf("OnNext: %d\n", v); },
	  []() { printf("OnCompleted\n"); });
  init.
	  subscribe(
	  [](int v) { printf("2nd OnNext: %d\n", v); },
	  []() { printf("2nd OnCompleted\n"); });
}

// buffer 收集Observable的数据放进一个数据包裹
void buffer() {
  auto values = rxcpp::observable<>::range(1, 10).buffer(3);
  values.
	  subscribe(
	  [](std::vector<int> v) {
		printf("OnNext:{");
		std::for_each(v.begin(), v.end(), [](int a) {
		  printf(" %d", a);
		});
		printf("}\n");
	  },
	  []() { printf("OnCompleted\n"); });
}

// 一个给定的延迟后发射一个特殊的值 timer操作符默认在computation调度器上执行
void timer() {
  auto scheduler = rxcpp::observe_on_new_thread();

  auto period = std::chrono::seconds(3);

  auto values = rxcpp::observable<>::timer(period, scheduler).
	  finally([]() {
	std::cout << "The final action, thread id: " << std::this_thread::get_id() << std::endl;
  });
  values.
	  as_blocking().
	  subscribe(
	  [](int v) { std::cout << "OnNext: " << v << "thread id: " << std::this_thread::get_id() << std::endl; },

	  []() { std::cout << "OnCompleted, thread id: " << std::this_thread::get_id() << std::endl; });
  std::cout << "main thread id: " << std::this_thread::get_id() << std::endl;
  //必须在主线程sleep，否则finally中的内容打印不全，因为程序退出后，线程销毁
  std::this_thread::sleep_for(std::chrono::milliseconds(100));
}

class Student {
 public:
  Student(std::string name, int id, std::string email)
	  : m_name(name), m_id(id), m_email(email) {}

  void printSelf() const {
	std::cout << "name: " << m_name << " "
			  << "id: " << m_id << " "
			  << "email: " << m_email << std::endl;
  }

 private:
  std::string m_name;
  int m_id;
  std::string m_email;
};

/**
 * @brief
 * @return
 */
int main(int argc, char **argv) {
  std::cout << "start info " << std::endl;
  testing::InitGoogleTest(&argc, argv);
  RUN_ALL_TESTS();
  system("pause");
}

// ! Demonstrate some basic assertions.
/**
 * * moads
 */
TEST(Coding, hello_world) {

  system("chcp 65001 > nul");
  printf("你看到了吗\n");

  // /------------ Print the Main Thread Id
  printf("Main Thread Id is %d\n",
		 std::this_thread::get_id());

  //------- Instantiate a run_loop object
  //------- which will loop in the main thread
  rxcpp::schedulers::run_loop rlp;

  //------ Create a coordination for run loop
  auto main_thread = rxcpp::observe_on_run_loop(rlp);
  auto worker_thread = rxcpp::synchronize_new_thread();

  rxcpp::composite_subscription scr;

  rxcpp::observable<>::range(0, 15).
	  map([](int i) {
	//----- This will get executed in worker
	printf("Map %d : %d\n", std::this_thread::get_id(), i);
	return i;
  }).
	  take(5).subscribe_on(worker_thread).
	  observe_on(main_thread).
	  subscribe(scr, [&](int i) {
	//--- This will get executed in main thread
	printf("Sub %d : %d\n", std::this_thread::get_id(), i);
  });

  //------------ Execute the Run Loop
  while (scr.is_subscribed() || !rlp.empty()) {
	while (!rlp.empty() && rlp.peek().when < rlp.now()) {
	  rlp.dispatch();
	}
  }

  folly::ConcurrentHashMap<std::string, Student> students;
  students.insert("Tom", Student("Tom", 1, "tom@gmail.com"));
  students.insert("Lilly", Student("Lilly", 2, "lilly@gmail.com"));

  for (const auto &st : students) {
	st.second.printSelf();
  }
  std::cout << "bar_ints = 🤣" << testing::PrintToString(students) << std::endl;

  std::vector<std::string> string{"a", "b", "c"};

  //std::cout << "bar_ints = " << testing::PrintToString(string) << std::endl;

  dbg(string);

  // Expect two strings not to be equal.
  EXPECT_STRNE("hello", "world");
  // Expect equality.
  EXPECT_EQ(7 * 6, 42);
}

// test
TEST(Coding, test_level_db) {
  using namespace std;
  leveldb::DB *db;
  leveldb::Options options;
  options.create_if_missing = true;

  // 打开数据库连接
  leveldb::Status status = leveldb::DB::Open(options, "./test_level_db", &db);
  assert(status.ok());
  string key = "weather";
  string value = "clearday";
  // 写入数据
  status = db->Put(leveldb::WriteOptions(), key, value);
  assert(status.ok());
  // 读取数据
  status = db->Get(leveldb::ReadOptions(), key, &value);
  assert(status.ok());
  cout << "value :" << value << endl;
  // 删除数据
  status = db->Delete(leveldb::WriteOptions(), key);
  assert(status.ok());
  status = db->Get(leveldb::ReadOptions(), key, &value);
  if (!status.ok()) {
	cerr << key << "   " << status.ToString() << endl;
  } else {
	cout << key << "===" << value << endl;
  }
  delete db;
}


